package com.atguigu.flink.chapter06.state;

import com.atguigu.flink.function.WaterSensorMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
import org.apache.flink.runtime.state.storage.JobManagerCheckpointStorage;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import sun.security.x509.GeneralName;

import java.io.IOException;

/*
状态后端（State Backends）
    哈希表状态后端（HashMapStateBackend）
    内嵌RocksDB状态后端（EmbeddedRocksDBStateBackend）
 */
public class StateBackendDemo {
    public static void main(String[] args) throws IOException {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);


        //设置状态后端
        //env.setStateBackend(new HashMapStateBackend());   //设置 HashMap状态后端
        //env.setStateBackend(new EmbeddedRocksDBStateBackend());   //设置内嵌 RocksDb状态后端
        env.setStateBackend(new EmbeddedRocksDBStateBackend(true));   //设置内嵌 RocksDb状态后端
        //远程状态存储（Checkpoint 存储的地方）
        //env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage());   //Chechpoint 存储JobManager内存
        //env.getCheckpointConfig().setCheckpointStorage("hdfs://......");   //Checkpoint存储到hdfs



        //旧的写法：
        //1、本地在TaskManager 的内存   2、远程在 JobManager内存
        env.setStateBackend(new MemoryStateBackend());

        //1、本地在TaskManager 的内存  2、远程在 Hdfs
        env.setStateBackend(new FsStateBackend("hdfs/....."));

        //1、本地在 RocksDb  2、远程在 Hdfs
        env.setStateBackend(new RocksDBStateBackend("hdfs:/"));


        env.socketTextStream("hadoop102",8888)
                .map(new WaterSensorMapFunction())
                .print();
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}



//public class StateBackendDemo {
//    public static void main(String[] args) throws IOException {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port", 2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(2);
//
//        //设置状态后端
//        //env.setStateBackend(new HashMapStateBackend());     //设置HashMap状态后端
//        //env.setStateBackend(new EmbeddedRocksDBStateBackend()) //设置内嵌的 RocksDB 状态后端
//        env.setStateBackend(new EmbeddedRocksDBStateBackend(true));
//        //远程状态存储（Checkpoint存储的地方）
//        //env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage());  //存储在  JObManager 的内存
//        //env.getCheckpointConfig().setCheckpointStorage("hdfs:/............");   //存储在 hdfs
//
//        //旧的方法
//        //1、本地在 TaskManager 的内存   2、远程在 JobManager 的内存
//        env.setStateBackend(new MemoryStateBackend());
//
//        //1、本地在 TaskManager 的内存   2、远程在 hdfs
//        env.setStateBackend(new FsStateBackend("hdfs:/....."));
//
//        //1、本地在 RocksDb       2、远程在 hdfs
//        env.setStateBackend(new RocksDBStateBackend("hdfs:/.........."));
//
//        env.socketTextStream("hadoop102",8888)
//                .map(new WaterSensorMapFunction())
//                .print();
//
//
//        try {
//            env.execute();
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }
//}


//public class StateBackendDemo {
//    public static void main(String[] args) throws IOException {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port", 2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(2);
//
//        //设置状态后端
//        //env.setStateBackend(new HashMapStateBackend());   //设置HashMap 状态后端
//        //env.setStateBackend(new EmbeddedRocksDBStateBackend());  //设置 内嵌的RocksDB 状态后端
//        env.setStateBackend(new EmbeddedRocksDBStateBackend(true));
//        //远程状态存储（Checkpoint存储的地方）
//        //env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage());  //存储到 JobManager 的内存
//        //env.getCheckpointConfig().setCheckpointStorage("hdfs:/.........");   //存储到 hdfs
//
//        //旧的方法
//        //1、本地在 TaskManager 的内存  2、远程在 JobManager 的内存
//        env.setStateBackend(new MemoryStateBackend());
//
//        //1、本地在 TaskManager 的内存  2、远程在 hdfs
//        env.setStateBackend(new FsStateBackend("hdfs://......"));
//
//        //1、本地在 RocksDb  2、远程在 hdfs
//         env.setStateBackend(new RocksDBStateBackend("hdfs://....."));
//
//        env.socketTextStream("hadoop102",8888)
//                .map(new WaterSensorMapFunction())
//                .print();
//
//        try {
//            env.execute();
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }
//}


//public class StateBackendDemo {
//    public static void main(String[] args) throws IOException {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port", 2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(2);
//
//        //设置状态后端
//        //env.setStateBackend(new HashMapStateBackend());  //设置HashMap状态后端
//        //env.setStateBackend(new EmbeddedRocksDBStateBackend()); //设置内嵌的 RocksDB 状态后端
//        env.setStateBackend(new EmbeddedRocksDBStateBackend());
//        //远程状态存储（Checkpoint存储的地方）
//        //env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage());  //存储在 JobManager 的内存
//        //env.getCheckpointConfig().setCheckpointStorage("hdfs://........");    //存储在hdfs
//
//        //旧的方法
//        //1、本地存储在 TaskManager 的内存  2、远程在  JobManager 的内存
//        env.setStateBackend(new MemoryStateBackend());
//
//        //1、本地存储在 TaskManager 的内存  2、远程在 hdfs
//        env.setStateBackend(new FsStateBackend("hdfs:/....."));
//
//        //1、本地存储在 RocksDB  2、远程在 hdfs
//        env.setStateBackend(new RocksDBStateBackend("hdfs:/..."));
//
//
//        try {
//            env.execute();
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }
//}